package cn.territory


import java.sql.{Connection, DriverManager}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import java.sql.{Connection, DriverManager}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object territoryV2 {

  def main(args: Array[String]): Unit = {
    if(args.length!=1){
      println(
        """
          |参数：
          |inputpath
          |""".stripMargin)
      sys.exit()
    }
    val Array(inputpath) = args

    val conf: SparkConf = new SparkConf().setAppName("territoryV2").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    val file: RDD[String] = sc.textFile(inputpath)

    val RegionRdd: RDD[((String, String), (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = file
      .map(line => line.split(",", line.length)).filter(_.length == 85).map(lines => {

      val adorderid: Int = lines(2).toInt
      val adcreativedid: Int = lines(3).toInt
      val applatformproviderid: Int = lines(4).toInt
      val requestmode: Int = lines(8).toInt
      val processnode: Int = lines(35).toInt
      val iseffective: Int = lines(30).toInt
      val isbilling: Int = lines(31).toInt
      val isbid: Int = lines(39).toInt
      val iswin: Int = lines(42).toInt
      val provincename = lines(24)
      val cityname = lines(25)
      val winprice = lines(41).toDouble
      val adpayment = lines(78).toDouble

      val originlRequest = territoryUtil.originlRequest(processnode, requestmode)
      val valueRequest = territoryUtil.valueRequest(processnode, requestmode)
      val advertiseRequest = territoryUtil.valueRequest(processnode, requestmode)
      val biddingTime = territoryUtil.biddingTime(iseffective, isbilling, isbid, adorderid)
      val biddingSucess = territoryUtil.biddingSucess(iseffective, isbilling, iswin)
      val terminalShow = territoryUtil.terminalShow(iseffective, processnode)
      val audienceClick = territoryUtil.audienceClick(iseffective, processnode)
      val DSPAdvertiseConsuem = territoryUtil.DSPAdvertiseConsuem(iseffective, isbilling, iswin, winprice)
      val DSPAdvertiseCos = territoryUtil.DSPAdvertiseConsuem(iseffective, isbilling, iswin, adpayment)

      ((provincename, cityname), (originlRequest, valueRequest, advertiseRequest, biddingTime, biddingSucess, terminalShow
        , audienceClick, DSPAdvertiseConsuem, DSPAdvertiseCos))
    })
    val key: RDD[((String, String), (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = RegionRdd.reduceByKey((x, y) => {
      (x._1 + y._1,
        x._2 + y._2,
        x._3 + y._3,
        x._4 + y._4,
        x._5 + y._5,
        x._6 + y._6,
        x._7 + y._7,
        x._8 + y._8,
        x._9 + y._9)
    })
    println(key.collect().toBuffer)

    key.foreachPartition(lines=>{
      lines.foreach(line=>{
        val conn:Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/lulala?characterEncoding=utf-8",
          "root", "123456")
        val ps = conn.prepareStatement("insert into rpt_pc_territory_V2 values(?,?,?,?,?,?,?,?,?,?,?)")
        ps.setString(1,line._1._1)
        ps.setString(2,line._1._1)
        ps.setInt(3,line._2._1)
        ps.setInt(4,line._2._2)
        ps.setInt(5,line._2._3)
        ps.setInt(6,line._2._4)
        ps.setInt(7,line._2._5)
        ps.setInt(8,line._2._6)
        ps.setInt(9,line._2._7)
        ps.setDouble(10,line._2._8)
        ps.setDouble(11,line._2._9)
        ps.executeUpdate()

        ps.close()
        conn.close()

      })
    })


  }

}
